package com.shujia.wyh.hadoop_hdfs;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;

import java.io.*;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;

/*
    我们学习java操作HDFS的目的主要是今后在程序中，
    使用代码对HDFS做一些操作：创建文件夹，上传文件，下载文件，查看目录，查看block块的信息，权限，大小，时间等等
 */
public class HDFSApiTest {
    public static void main(String[] args) throws Exception {
        //加载hadoop的配置文件对象
        Configuration conf = new Configuration();
        //windows作为客户端远程连接hadoop集群，需要指定hadoop中的主节点的信息
        conf.set("fs.defaultFS","hdfs://master:9000");

        //hadoop2.x之后提供了一个抽象类FileSystem类，这个类中有一个静态方法，可以让我们获取对应的连接对象
        //之后我们在拿着这个连接对象就可以操作我们的hadoop集群了
        FileSystem fs = FileSystem.get(conf);
        System.out.println(fs);

        //尝试在HDFS中创建文件夹
//        boolean b = fs.mkdirs(new Path("/shujia/test4"));
//        System.out.println(b);

        //上传文件：
        //第一种方式：调用fs中的方法进行创建
        //第一个参数代表的是windows中的本地路径，第二个参数代表的是HDFS中的路径
        fs.copyFromLocalFile(
                new Path("bigdata23-hadoop/data/students.txt"),
                new Path("/shujia/test4/"));

        //下载文件
        //第一个参数代表的是HDFS的路径，第二个参数代表的是Windows的路径
        fs.copyToLocalFile(
                new Path("/shujia/test4/students.txt"),
                new Path("bigdata23-hadoop/data/students2.txt"));

        //列出当前目录下的所有文件
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/shujia"), true);
        while (iterator.hasNext()){
            //获取到了一个一个文件
            LocatedFileStatus fileStatus = iterator.next();
            //获取所有的block块
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String name = Arrays.toString(blockLocation.getNames());
                System.out.println(name);
            }

            //获取文件的所属用户信息
            String owner = fileStatus.getOwner();
            System.out.println(owner);

            //获取文件的名字,间接获取文件的名字，先获取文件的路径，然后就能获取文件名称了
            String name = fileStatus.getPath().getName();
            System.out.println(name);

            //获取文件的权限
            FsPermission permission = fileStatus.getPermission();
            String s = permission.toString();
            System.out.println(s);

            //获取文件最后修改的时间
            long time = fileStatus.getModificationTime();
//            System.out.println(time);
            Date date = new Date(time);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String format = sdf.format(date);
            System.out.println(format);

            System.out.println("==========================================");
        }


        //第二种上传方式：使用对流复制的方式
        //创建输入流对象
        BufferedReader br = new BufferedReader(new FileReader("bigdata23-hadoop/data/students.txt"));
        //创建输出流对象
        //使用HDFS的文件系统对象获取输出流对象
        FSDataOutputStream fsDataOutputStream = fs.create(new Path("/shujia/test4/b.txt"));
        IOUtils.copy(br,fsDataOutputStream);
        //释放资源
        IOUtils.closeQuietly(fsDataOutputStream);
        IOUtils.closeQuietly(br);


        //使用对流复制的方式进行下载
        //使用HDFS的文件系统对象获取输入流对象
        FSDataInputStream fsDataInputStream = fs.open(new Path("/shujia/test4/b.txt"));
        //创建本地的输出流对象
        BufferedWriter bw = new BufferedWriter(new FileWriter("E:\\projects\\IDEAProjects\\bigdata23-project\\bigdata23-project\\bigdata23-hadoop\\data\\students5.txt"));
        IOUtils.copy(fsDataInputStream,bw);


        //释放资源
        IOUtils.closeQuietly(bw);
        IOUtils.closeQuietly(fsDataInputStream);
        fs.close();
    }
}
